package com.pxene.dmp.mapper;

import java.io.IOException;
import java.net.MalformedURLException;
import java.net.URL;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import com.pxene.dmp.common.URLUtils;
import com.pxene.dmp.constant.PatternList;
import com.pxene.dmp.constant.PatternList.Param;
import com.pxene.dmp.constant.PatternList.Pattern;

public class GetBusinessCodeMapper extends Mapper<LongWritable, Text, Text, Text> {

	private URL urlParser;
	private Text resultKey = new Text();
	private Text resultVal = new Text();

	@Override
	protected void setup(Context context) throws IOException, InterruptedException {
		Configuration conf = context.getConfiguration();
		// 设置配置文件名
		System.setProperty("config-file", conf.get("config-file"));
	}

	@Override
	protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
		String[] columns = value.toString().split("\\|");
		if (columns.length != 7) {
			return;
		}
		String imsi = columns[0];
		String bsid = columns[1];
		String starttime = columns[2];
		String name = columns[3];
		String url = columns[4];
		String province = columns[5];
		String dataType = columns[6];

		try {
			urlParser = new URL(url);
		} catch (MalformedURLException e) {
			return;
		}

		String code = "";
		Map<String, String> map = URLUtils.getQueryMap(urlParser.getQuery());

		if (code.length() == 0) {
			PatternList businessCodeList = PatternList.load("businesscode.json");
			for (Pattern businessCode : businessCodeList.elements()) {
				if (businessCode.getCode().equals(name)) {
					String expr = businessCode.getParams()[0].getExpr();
					code = URLUtils.regexpExtract(url, expr).trim();
					break;
				}
			}
		}

		if (code.length() == 0) {
			PatternList contactNumList = PatternList.load("contactnum.json");
			for (Pattern contactNum : contactNumList.elements()) {
				if (contactNum.getCode().equals(name)) {
					String expr = contactNum.getParams()[0].getExpr();
					code = URLUtils.regexpExtract(url, expr).trim();
					break;
				}
			}
		}

		if (code.length() == 0) {
			PatternList locationList = PatternList.load("location.json");
			for (Pattern location : locationList.elements()) {
				if (location.getCode().equals(name)) {
					Param[] params = location.getParams();
					if (params.length == 2) {
						String expr0 = params[0].getExpr();
						String expr1 = params[1].getExpr();
						String lng = "";
						String lat = "";
						lng = URLUtils.regexpExtract(url, expr0).trim();
						lat = URLUtils.regexpExtract(url, expr1).trim();
						code = lng + "," + lat;
					} else {
						String expr = location.getParams()[0].getExpr();
						code = URLUtils.regexpExtract(url, expr).trim();
					}
					break;
				}
			}
		}

		if (code.length() > 0) {
			resultKey.set(imsi + "|" + bsid + "|" + starttime + "|" + name + "|" + province + "|" + dataType);
			resultVal.set(code);
			context.write(resultKey, resultVal);
		}
	}
}
